#ifndef AVALON_SERVICENODE_HPP
#define AVALON_SERVICENODE_HPP
#include <vector>
#include <boost/call_traits.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/preprocessor/repetition/enum_params.hpp>
#include <boost/preprocessor/repetition/enum_binary_params.hpp>
#include <boost/preprocessor/repetition/repeat.hpp>
#include <boost/preprocessor/iteration/local.hpp>
#include "Avalon/IO/Buffer.hpp"
#include "Avalon/IO/Channel.hpp"
#include "Avalon/IO/Connection.hpp"
#include "Avalon/IO/ConnectionState.hpp"
#include "Avalon/IO/NotConnectedException.hpp"
#include "Avalon/IO/Writer.hpp"
#include "Avalon/Serialization/Deserializer.hpp"
#include "Avalon/Serialization/Serializer.hpp"
#include "Avalon/Services/Message.hpp"
#include "Avalon/SignalHandling/TaskSignalHandler.hpp"
#include "Avalon/Threading/Async.hpp"
#include "Avalon/Threading/Timer.hpp"

namespace Avalon {
namespace Services {

  /*! \class ServiceNode
   *  \brief Represents a single node on a service network.
   *  \author Kamal Mansouri <kranar@hotmail.com>
   */
  class ServiceNode {
    public:

      //! The default timeout for a Channel.
      static const boost::posix_time::time_duration TIMEOUT;

      //! Constructs a ServiceNode.
      /*!
        \param dataShuttleFactory The type of Message Serialization to use.
        \param timeout The amount of time to wait for a heartbeat.
      */
      ServiceNode(Serialization::DataShuttleFactory* dataShuttleFactory,
        boost::posix_time::time_duration timeout);

      ~ServiceNode();

      //! Returns the signal for a specified Message.
      template<typename T>
      typename T::Signal& GetSignal() {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        std::map<const std::type_info*,
          boost::signals2::signal_base*>::iterator signalIterator =
          m_signals.find(&typeid(T));
        if(signalIterator == m_signals.end()) {
          signalIterator = m_signals.insert(make_pair(&typeid(T),
            new typename T::Signal())).first;
        }
        return *(static_cast<typename T::Signal*>(signalIterator->second));
      }

      //! Sends a Message to a Channel.
      /*!
        \param channel The Channel to send the <i>message</i> to.
        \param message The Message to send.
      */
      void SendMessage(IO::Channel* channel, const Message& message);

      //! Sends a Message to a collection of Channels.
      /*!
        \param channels The Channels to send the <i>message</i> to.
        \param message The Message to send.
      */
      template<typename T>
      void SendMessage(const std::vector<T>& channels, const Message& message) {
        for(typename std::vector<T>::const_iterator i = channels.begin();
            i != channels.end(); ++i) {
          SendMessage(*i, message);
        }
      }

      //! Sends a Service request.
      /*!
        \param channel The ServiceNode Channel to send the request to.
        \return The result of the Service request.
      */
      template<typename T>
      typename Threading::Async<typename T::ReturnType>::Ptr SendRequest(
          IO::Channel* channel) {
        typedef typename T::Request Request;
        typedef Threading::AsyncOwner<typename T::ReturnType> AsyncResponse;
        boost::lock_guard<boost::mutex> lock(m_mutex);
        std::map<IO::Channel*, ChannelEntry*>::iterator channelIterator =
          m_channelEntries.find(channel);
        assert(channelIterator != m_channelEntries.end());
        ChannelEntry* entry = channelIterator->second;
        if(!entry->m_state.IsOpen()) {
          BOOST_THROW_EXCEPTION(IO::NotConnectedException());
        }
        int requestId = m_nextRequestId;
        ++m_nextRequestId;
        Request request(requestId);
        entry->m_serializer->Shuttle("message", &request);
        channel->GetWriter().Write(entry->m_serializer->GetSerializedData());
        typename AsyncResponse::Ptr response(AsyncResponse::Create());
        m_pendingRequests.insert(std::make_pair(requestId, response));
        entry->m_responses.push_back(response);
        return response;
      }

      #ifndef AVALON_SERVICENODE_SEND_REQUEST_CALL_ARGUMENTS
        #define AVALON_SERVICENODE_SEND_REQUEST_CALL_ARGUMENTS 20
      #endif

      #define AVALON_SERVICENODE_DEFINE_PARAMETER(z, n, q)                     \
        BOOST_PP_COMMA_IF(n) typename boost::call_traits<                      \
          typename T::Parameter##n::Type>::param_type a##n

      #define AVALON_SERVICENODE_SEND_REQUEST_CALL_COMMA_COPY(z, n, a)         \
        BOOST_PP_COMMA_IF(n) a##n

      #define BOOST_PP_LOCAL_MACRO(n)                                          \
      template<typename T>                                                     \
        typename Threading::Async<typename T::ReturnType>::Ptr                 \
           SendRequest(IO::Channel* channel,                                   \
           BOOST_PP_REPEAT(n, AVALON_SERVICENODE_DEFINE_PARAMETER, A)) {       \
        typedef typename T::Request Request;                                   \
        typedef Threading::AsyncOwner<typename T::ReturnType> AsyncResponse;   \
        boost::lock_guard<boost::mutex> lock(m_mutex);                         \
        std::map<IO::Channel*, ChannelEntry*>::iterator channelIterator =      \
          m_channelEntries.find(channel);                                      \
        assert(channelIterator != m_channelEntries.end());                     \
        ChannelEntry* entry = channelIterator->second;                         \
        if(!entry->m_state.IsOpen()) {                                         \
          BOOST_THROW_EXCEPTION(IO::NotConnectedException());                  \
        }                                                                      \
        int requestId = m_nextRequestId;                                       \
        ++m_nextRequestId;                                                     \
        Request request(requestId, BOOST_PP_REPEAT(n,                          \
          AVALON_SERVICENODE_SEND_REQUEST_CALL_COMMA_COPY, a));                \
        entry->m_serializer->Shuttle("message", &request);                     \
        channel->GetWriter().Write(entry->m_serializer->GetSerializedData());  \
        typename AsyncResponse::Ptr response(AsyncResponse::Create());         \
        m_pendingRequests.insert(std::make_pair(requestId, response));         \
        entry->m_responses.push_back(response);                                \
        return response;                                                       \
      }

      #define BOOST_PP_LOCAL_LIMITS                                            \
        (1, AVALON_SERVICENODE_SEND_REQUEST_CALL_ARGUMENTS)
      #include BOOST_PP_LOCAL_ITERATE()

      //! Adds a ServiceNode Channel to the network.
      /*!
        \param channel The ServiceNode Channel to add.
        \param slot The slot receiving the ClosedSignal notification.
        \param isConnected [out] <code>true</code> iff the <i>channel</i>'s
                           connection is open.
        \returns The ClosedSignal indicating that the Channel has disconnected.
      */
      boost::signals2::connection AddChannel(IO::Channel* channel,
        const IO::Connection::ClosedSignal::slot_type& slot, bool* isConnected);

      //! Removes a ServiceNode Channel from the network.
      /*!
        \param channel The ServiceNode Channel to remove.
      */
      void RemoveChannel(IO::Channel* channel);

      //! Closes a ServiceNode Channel and removes it from the network.
      /*!
        \param channel The ServiceNode Channel to remove.
      */
      void CloseAndRemoveChannel(IO::Channel* channel);

    private:
      struct ChannelEntry {
        IO::Channel* m_channel;
        boost::scoped_ptr<Serialization::Serializer> m_serializer;
        boost::scoped_ptr<Serialization::Deserializer> m_deserializer;
        IO::ConnectionState m_state;
        bool m_sentHeartbeat;
        int m_timeoutCounter;
        IO::Buffer m_data;
        bool m_isReadPending;
        boost::condition_variable m_closedCondition;
        IO::Connection::ClosedSignal m_closedSignal;
        std::vector<Threading::BaseAsyncOwnerPtr> m_responses;
      };
      boost::mutex m_mutex;
      boost::scoped_ptr<Serialization::DataShuttleFactory>
        m_dataShuttleFactory;
      std::map<IO::Channel*, ChannelEntry*> m_channelEntries;
      std::map<int, Threading::BaseAsyncPtr> m_pendingRequests;
      std::map<const std::type_info*, boost::signals2::signal_base*> m_signals;
      Threading::Timer m_heartbeatTimer;
      int m_nextRequestId;
      SignalHandling::TaskSignalHandler m_signalHandler;

      void InternalRemoveChannel(
        std::map<IO::Channel*, ChannelEntry*>::iterator& channelIterator);
      void OnClosed(ChannelEntry* channel);
      void OnRead(const Threading::Async<int>::Ptr& result,
        ChannelEntry* entry);
      void OnHeartbeat();
  };

  /*! \class RequestToken
   *  \brief Represents a ServiceNode request.
   *  \author Kamal Mansouri <kranar@hotmail.com>
   */
  template<typename Service>
  class RequestToken {
    public:

      //! Constructs a RequestToken.
      /*!
        \param serviceNode The ServiceNode that received the request.
        \param channel The Channel the request was sent through.
        \param requestId The request id.
      */
      RequestToken(ServiceNode* serviceNode, IO::Channel* channel,
          int requestId)
          : m_serviceNode(serviceNode),
            m_channel(channel),
            m_requestId(requestId) {}

      ~RequestToken() {}

      //! Returns the Channel the request was sent through.
      IO::Channel* GetChannel() const {
        return m_channel;
      }

      //! Sends a response to this request.
      /*!
        \param response The response to send to this request.
      */
      void SendResponse(const typename Service::ReturnType& response) const {
        m_serviceNode->SendMessage(m_channel, typename Service::Response(
          m_requestId, response));
      }

    private:
      ServiceNode* m_serviceNode;
      IO::Channel* m_channel;
      int m_requestId;
  };

  //! Sets the slot for a Message.
  /*!
    \param serviceNode The ServiceNode emitting the signal.
    \param signalHandler The signal handler receiving the message.
    \param slot The slot to connect to the Message signal.
  */
  template<typename T, typename S>
  void SetMessageHandler(ServiceNode& serviceNode, S& signalHandler,
      const typename T::Signal::slot_type& slot) {
    serviceNode.template GetSignal<T>().connect(
      signalHandler.template GetSlot<typename T::Signal>(slot));
  }

  //! Sets the slot for a RequestMessage.
  /*!
    \param serviceNode The ServiceNode emitting the signal.
    \param signalHandler The signal handler receiving the message.
    \param slot The slot to connect to the RequestMessage signal.
  */
  template<typename T, typename S>
  void SetRequestHandler(ServiceNode& serviceNode, S& signalHandler,
      const typename T::Request::Signal::slot_type& slot) {
    serviceNode.template GetSignal<typename T::Request>().connect(
      signalHandler.template GetSlot<typename T::Request::Signal>(slot));
  }
}
}

#endif // AVALON_SERVICENODE_HPP
